package com.seari.client.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Arrays;
import java.util.Properties;

public class Test extends Thread{

    public KafkaConsumer<String, String> consumer;
    private static final String topic = "LINE1_ATS|LINE2_ATS|LINE3_ATS";
    private static final String bootstrapServers = "172.19.122.139:9092,172.19.122.140:9092,172.19.122.141:9092";
    private static final String groupId = "seari";
    private static final String enableAutoCommit = "true";
    private static final String autoOffsetReset = "latest";
    private static final int autoCommitIntervalMs = 1000;

    public Test() {
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServers);
        props.put("group.id", groupId);
        props.put("enable.auto.commit", enableAutoCommit);
        props.put("auto.commit.interval.ms", autoCommitIntervalMs);
        props.put("auto.offset.reset", autoOffsetReset);
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        this.consumer = new KafkaConsumer<String, String>(props);
        this.consumer.subscribe(Arrays.asList(topic.split("\\|")));
    }
        @Override
		public void run() {
			  System.out.println(bootstrapServers);
		      while (true) {

		            ConsumerRecords<String, String> records = consumer.poll(100);

                  for (ConsumerRecord<String, String> record : records) {
		                /** print the offset,key and value for the consumer records.打印消费者记录的偏移量、键和值。*/
		                System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());
		            }
		      }
		}

    public static void main(String args[]) {
			Test test1 = new Test();
			Thread thread1 = new Thread(test1);
			thread1.start();
    }
}
